package com.ljm.boot.kafka.consumer;

import com.ljm.boot.kafka.config.KafkaConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @Description
 * @Author Dominick Li
 * @CreateTime 2024/1/16 14:38
 **/
@Slf4j
@Component
public class KafkaConsumer {

    /**
     * 监听消息
     */
    @KafkaListener(topics = {KafkaConfig.DEFALUT_TOPIC}, concurrency = "5")
    public void onMessage(ConsumerRecord<String, Object> record) {
        log.info("onMessage msg={}",record.value());
    }

    /**
     * 同一主题批量消费groupId不能和单次消费的一样
     */
//    @KafkaListener(topics = {KafkaConfig.DEFALUT_TOPIC}, errorHandler = KafkaConfig.CONSUMER_LISTENER_ERROR_HANDLER_NAME,groupId = "batchGroup")
//    public void onBatchMessage(List<ConsumerRecord<String, Object>> records) throws Exception {
//        log.info("batch size={}", records.size());
//        for (ConsumerRecord<String, Object> record : records) {
//            log.info("onBatchMessage msg={}", record.value());
//        }
//        //模拟异常
//        //throw  new Exception("test errorHandler");
//    }




}
